package com.aotain.project.apollo.bolt;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.common.CommonFunction;
import com.aotain.mushroom.Slave;
import com.aotain.storm.TupleHelpers;
import com.aotain.apollo.IDCIPSegment;
import com.aotain.apollo.IPDatabase;
import com.aotain.apollo.IPSearch;

/**
 * 
 *  @ClassName: SourceFmtBolt 
 * @Description: TODO(这里用一句话描述这个类的作用) 
 * @author 程彬
 * @date 2015年8月7日 下午2:56:10 
 *
 */
public class SourceFmtBolt implements IRichBolt {

	/** 
	 * @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	 */ 
	private static final long serialVersionUID = 7816556443307732593L;

	Slave s = null;
	String sourceAreaCountry = null;
	String sourceAreaCity = null;
	String sourceAreaProvince = null;
	String sourceAreaName = null;
	String destAreaCountry = null;   
	String destAreaCity = null;
	String destAreaName = null;
	String destAreaProvince = null;
	String gisstr = null;
	String gisstrBase64 = null;
	String Sourcelon = null;
	String Sourcelat = null;
	String Destlon = null;
	String Destlat = null;
	String sourceAreaId = null;
	String sourceAreaProvinceId = null;
	String sourceAreaCityId = null;
	String destAreaId = null;
	String destAreaProvinceId = null;
	String destAreaCityId = null;
	String areaName_s = null;
	String areaName_d = null;
	OutputCollector collector = null;
	HashMap<Long, IPDatabase> map = null;
	Long[] IPs = null;
	String dip = null;
	String dport = null;
	String sip = null;
	String sport = null;
	String filterIPs = null;
	String domainname = null;
	String url = null;
	String type = null;
	String accesstime = null;
	String idcno = null;
	String delaytime = null;
	//	ApolloConfig ap = null;
	IDCIPSegment _ipseg = null;
	IPSearch ipsearch = null;
	String logserver = null;
	String zooserver = null;
	HashSet<String> IPSet = new HashSet<String>();

	public SourceFmtBolt(){}

	public SourceFmtBolt(String filterIPs) {
		this.filterIPs = filterIPs;
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declareStream("DipStreamId",new Fields("dip","dport","sip","sport","gisstr","domainname","url","type","delaytime","idcno"));
		declarer.declareStream("SipStreamId",new Fields("dip","dport","sip","sport","gisstr","domainname","url","type","delaytime","idcno"));
	}   

	@Override
	public Map<String, Object> getComponentConfiguration() {
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,30*60);
		return conf;  
	}

	@Override
	public void execute(Tuple tuple) {
		try {
			if(TupleHelpers.isTickTuple(tuple)) {
				emitCountingData();  	
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println(" AbnStatisBolt exception"+e.getMessage());
			Logger.getRootLogger().error("AbnStatisBolt exception====",e);
			collector.fail(tuple);
			throw new FailedException(e	);
		}	
	}

	private void countInLocal(Tuple tuple) {
		// TODO Auto-generated method stub
		String logString = tuple.getString(0);
		if(logString != null && !"".equals(logString)) {
			String[] splits = logString.split("\\|",-1);
			if(splits.length>9) {
				dip = splits[2];
				dport = splits[5];
				sip = splits[1];
				sport = splits[4];
				domainname = splits[6];
				url = splits[7];
				type = splits[3]; 
				accesstime= splits[9];
				idcno = splits[0];

				delaytime = 0 + "";
				long currenttime = System.currentTimeMillis();
				if(accesstime != null && !"".equals(accesstime)) {
					delaytime = currenttime/1000 - Long.parseLong(accesstime.trim()) + "";
				}
				if(IPSet == null || IPSet.contains(dip)) {
					
					//				boolean bl = _ipseg.IsLegal(idcno, dip);
					/**合并后
				if(_ipseg.IsLegal(idcno, dip) || _ipseg.IsLegal(idcno, sip)) {

					//$$$$$$测试
					if(!logserver.isEmpty() && dip.equals("183.60.118.212") && idcno.equals("DX-GZ-002-RMZ"))
					{
						Slave s = new Slave(logserver.split(":",-1)[0],
								Integer.parseInt(logserver.split(":",-1)[1]));
						String msg = String.format("SourceFmtBolt 's flter IP :%s ;IDCNO:%s", dip, idcno);
						s.SendAlarmLog("INFO", msg);
					}

					long currenttime = System.currentTimeMillis();
					if(accesstime != null && !"".equals(accesstime)) {
						delaytime = currenttime/1000 - Long.parseLong(accesstime.trim()) + "";
					} else {
						delaytime = 0 + "";
					}

					if(!"".equals(dip) && !"".equals(dport) && !"".equals(sip) && !"".equals(sport) ) {

						//						IPDatabase SourceArea = ApolloConfig.getStartIP(map,IPs,sip);
						//						IPDatabase DestArea = ApolloConfig.getStartIP(map,IPs,dip);
						IPDatabase SourceArea = ipsearch.getStartIP(sip);
						IPDatabase DestArea = ipsearch.getStartIP(dip);

						if(SourceArea != null) {
							sourceAreaCountry = SourceArea.getCountryName();
							sourceAreaCity = SourceArea.getCityName();
							//						sourceAreaProvince = SourceArea.getProviceName();
							if(sourceAreaCity != null) {
								sourceAreaName  = sourceAreaCity;
								sourceAreaId = String.valueOf(SourceArea.getCityID());
							} else {
								sourceAreaName = sourceAreaCountry;
								sourceAreaId = String.valueOf(SourceArea.getCountryID());
							}
							sourceAreaCityId = sourceAreaId;
							Sourcelon = String.valueOf(SourceArea.getLon());//经度
							Sourcelat = String.valueOf(SourceArea.getLat());//纬度

							if(!"".equals(SourceArea.getProviceID()) && SourceArea.getProviceID() !=0 ) {
								sourceAreaId = String.valueOf(SourceArea.getProviceID());
								areaName_s = SourceArea.getProviceName().replace("省", "");
								areaName_s = areaName_s.replace("市", "");
							} else {
								areaName_s = SourceArea.getCountryName();
								sourceAreaId = String.valueOf(SourceArea.getCountryID());
							}
							sourceAreaProvinceId = sourceAreaId;
						}   

						if(DestArea != null ) {    
							destAreaCountry = DestArea.getCountryName();
							destAreaCity = DestArea.getCityName();
							if(destAreaCity != null) {
								destAreaName = destAreaCity;
								destAreaId = String.valueOf(DestArea.getCityID());
							} else {   
								destAreaName = destAreaCountry;
								destAreaId = String.valueOf(DestArea.getCountryID());
							}
							destAreaCityId = destAreaId;
							Destlon = String.valueOf(DestArea.getLon());
							Destlat = String.valueOf(DestArea.getLat());

							if(!"".equals(DestArea.getProviceID()) && DestArea.getProviceID() !=0 ) {
								destAreaId = String.valueOf(DestArea.getProviceID());
								areaName_d = DestArea.getProviceName().replace("省", "");
								areaName_d = areaName_d.replace("市", "");
							} else {
								areaName_d = DestArea.getCountryName();
								destAreaId = String.valueOf(DestArea.getCountryID());
							}
							destAreaProvinceId = destAreaId;
						}  

						if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null && Sourcelon != null && Sourcelat != null && sourceAreaCountry != null) {
							//sourceAreaId destAreaId 均为省id
							gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry + "#" + sourceAreaId + "#" +areaName_s + "#" + sourceAreaCityId + "#" + sourceAreaProvinceId + "#" + sourceAreaCountry;
							gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
							collector.emit("DipStreamId",new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno));
							collector.ack(tuple);

							gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + destAreaCountry + "#" + destAreaId + "#" +areaName_d + "#" + destAreaCityId + "#" + destAreaProvinceId + "#" + sourceAreaCountry;
							gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
							collector.emit("SipStreamId",new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno));
							collector.ack(tuple);
						}

						//						if(DestArea != null) {
						//							destAreaCountry = DestArea.getCountryName();
						//							destAreaCity = DestArea.getCityName();
						//							//						sourceAreaProvince = SourceArea.getProviceName();
						//							if(destAreaCity != null) {
						//								destAreaName  = destAreaCity;
						//								destAreaId = String.valueOf(DestArea.getCityID());
						//							} else {
						//								destAreaName = destAreaCountry;
						//								destAreaId = String.valueOf(DestArea.getCountryID());
						//							}
						//							destAreaCityId = sourceAreaId;
						//							Destlon = String.valueOf(DestArea.getLon());
						//							Destlat = String.valueOf(DestArea.getLat());
						//
						//							if(!"".equals(DestArea.getProviceID()) && DestArea.getProviceID() !=0 ) {
						//								destAreaId = String.valueOf(DestArea.getProviceID());
						//								areaName_d = DestArea.getProviceName().replace("省", "");
						//								areaName_d = areaName_d.replace("市", "");
						//							} else {
						//								areaName_d = DestArea.getCountryName();
						//								destAreaId = String.valueOf(DestArea.getCountryID());
						//							}
						//							destAreaProvinceId = destAreaId;
						//						}   
						//
						//						if(SourceArea != null) {    
						//							sourceAreaCountry = SourceArea.getCountryName();
						//							sourceAreaCity = SourceArea.getCityName();
						//							if(sourceAreaCity != null) {
						//								sourceAreaName = sourceAreaCity;
						//							} else {   
						//								sourceAreaName = sourceAreaCountry;
						//							}
						//							Sourcelon = String.valueOf(SourceArea.getLon());//经度
						//							Sourcelat = String.valueOf(SourceArea.getLat());//纬度
						//						}  
						//
						//						if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null && Sourcelon != null && Sourcelat != null && sourceAreaCountry != null) {
						//							gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + destAreaCountry + "#" + destAreaId + "#" +areaName_d + "#" + destAreaCityId + "#" + destAreaProvinceId + "#" + sourceAreaCountry;
						//							gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
						////							collector.emit("SipStreamId",new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno));
						//							collector.ack(tuple);
						//						}
					}
				}
					 **/

					//					/** 合并两个emit 413
					//					if(!"".equals(dip) && !"".equals(dport) && !"".equals(sip) && !"".equals(sport) && IPSet.contains(dip)) {
					if(_ipseg.IsLegal(idcno, dip)) {
						if(!"".equals(dip) && !"".equals(dport) && !"".equals(sip) && !"".equals(sport) ) {
							url = CommonFunction.decodeBASE64(url);
							//						IPDatabase SourceArea = ApolloConfig.getStartIP(map,IPs,sip);
							//						IPDatabase DestArea = ApolloConfig.getStartIP(map,IPs,dip);
							IPDatabase SourceArea = ipsearch.getStartIP(sip);
							IPDatabase DestArea = ipsearch.getStartIP(dip);

							if(SourceArea != null) {
								sourceAreaCountry = SourceArea.getCountryName();
								sourceAreaCity = SourceArea.getCityName();
								//						sourceAreaProvince = SourceArea.getProviceName();
								if(sourceAreaCity != null) {
									sourceAreaName  = sourceAreaCity;
									sourceAreaId = SourceArea.getCityID()+"";
								} else {
									sourceAreaName = sourceAreaCountry;
									sourceAreaId = SourceArea.getCountryID()+"";
								}
								sourceAreaCityId = sourceAreaId;
								Sourcelon = String.valueOf(SourceArea.getLon());//经度
								Sourcelat = String.valueOf(SourceArea.getLat());//纬度

								if(!"".equals(SourceArea.getProviceID()) && SourceArea.getProviceID() !=0 ) {
									sourceAreaId = SourceArea.getProviceID()+"";
									areaName_s = SourceArea.getProviceName().replace("省", "");
									areaName_s = areaName_s.replace("市", "");
								} else {
									areaName_s = SourceArea.getCountryName();
									sourceAreaId = SourceArea.getCountryID()+"";
								}
								sourceAreaProvinceId = sourceAreaId;
							}   

							if(DestArea != null ) {    
								destAreaCountry = DestArea.getCountryName();
								destAreaCity = DestArea.getCityName();
								if(destAreaCity != null) {
									destAreaName = destAreaCity;
								} else {   
									destAreaName = destAreaCountry;
								}
								Destlon = String.valueOf(DestArea.getLon());
								Destlat = String.valueOf(DestArea.getLat());
							}  

							if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null
								&& Sourcelon != null && Sourcelat != null && sourceAreaCountry != null && sourceAreaId != null  && areaName_s != null
								) {
								gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#"
										+ Sourcelon + "," + Sourcelat + "#" + sourceAreaCountry + "#" + sourceAreaId + "#"
										+areaName_s + "#" + sourceAreaCityId + "#" + sourceAreaProvinceId + "#" + sourceAreaCountry;
								gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
								collector.emit("DipStreamId",new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno));
								collector.ack(tuple);
								return ;
							}

						}
					}
				}
				if(IPSet == null || IPSet.contains(sip)) {
					//					if(!"".equals(dip) && !"".equals(dport) && !"".equals(sip) && !"".equals(sport) && IPSet.contains(sip)) {
					if(_ipseg.IsLegal(idcno, sip)) {
						if(!"".equals(dip) && !"".equals(dport) && !"".equals(sip) && !"".equals(sport)) {
							url = CommonFunction.decodeBASE64(url);
							IPDatabase SourceArea = ipsearch.getStartIP(sip);
							IPDatabase DestArea = ipsearch.getStartIP(dip);

							if(DestArea != null) {
								destAreaCountry = DestArea.getCountryName();
								destAreaCity = DestArea.getCityName();
								//						sourceAreaProvince = SourceArea.getProviceName();
								if(destAreaCity != null) {
									destAreaName  = destAreaCity;
									destAreaId = DestArea.getCityID()+"";
								} else {
									destAreaName = destAreaCountry;
									destAreaId = DestArea.getCountryID()+"";
								}
								destAreaCityId = sourceAreaId;
								Destlon = String.valueOf(DestArea.getLon());
								Destlat = String.valueOf(DestArea.getLat());

								if(!"".equals(DestArea.getProviceID()) && DestArea.getProviceID() !=0 ) {
									destAreaId = DestArea.getProviceID()+"";
									areaName_d = DestArea.getProviceName().replace("省", "");
									areaName_d = areaName_d.replace("市", "");
								} else {
									areaName_d = DestArea.getCountryName();
									destAreaId = DestArea.getCountryID()+"";
								}
								destAreaProvinceId = destAreaId;
							}   

							if(SourceArea != null) {    
								sourceAreaCountry = SourceArea.getCountryName();
								sourceAreaCity = SourceArea.getCityName();
								if(sourceAreaCity != null) {
									sourceAreaName = sourceAreaCity;
								} else {   
									sourceAreaName = sourceAreaCountry;
								}
								Sourcelon = String.valueOf(SourceArea.getLon());//经度
								Sourcelat = String.valueOf(SourceArea.getLat());//纬度
							}  

							if(destAreaName != null && Destlon != null && Destlat != null && sourceAreaName != null && Sourcelon != null && Sourcelat != null && sourceAreaCountry != null && destAreaId != null  && areaName_d != null) {
								gisstr = destAreaName + "#" + Destlon + "," + Destlat + "#" + sourceAreaName + "#" + Sourcelon + "," + Sourcelat + "#" + destAreaCountry + "#" + destAreaId + "#" +areaName_d + "#" + destAreaCityId + "#" + destAreaProvinceId + "#" + sourceAreaCountry;
								gisstrBase64 = new String(Base64.encodeBase64(gisstr.getBytes()));
								collector.emit("SipStreamId",new Values(dip,dport,sip,sport,gisstrBase64,domainname,url,type,delaytime,idcno));
								collector.ack(tuple);
							}
						}
					}
					//					 **/
				}
			}
		}
	}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		if(!logserver.isEmpty())
		{
			Slave s = new Slave(zooserver);

			s.SendAlarmLog("ERROR", "SourceBolt Shutdown");
		}
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collector = collector;
		//		String ips = "101.227.130.26,180.163.28.22,101.227.160.27,61.151.186.3,182.254.150.24,182.254.139.240,203.195.193.178,182.254.219.60,182.254.225.173,61.188.37.186,182.254.241.239,115.159.30.208,222.73.33.114,222.73.33.115,222.73.33.116,101.226.182.53,101.226.182.92,222.73.155.185,101.226.182.75,222.73.195.213,183.60.170.217,101.226.64.227,115.159.26.254,101.226.131.205,101.226.69.253,183.62.140.141,113.116.229.21";
		//		ZkKafkaConfig zkConfig = new ZkKafkaConfig("/home/storm/config/dbconfig.ini");
		//		String ips = zkConfig.getFilterIPss();
		//		String[] ipsArray = filterIPs.split(",");
		//		IPSet = new HashSet<String>(Arrays.asList(ipsArray));//数组转为set

		String dbJson = stormConf.get("dbjson").toString();
		logserver = stormConf.get("mushroomserver").toString();
		//		ap = new ApolloConfig(dbJson, mushroomServer);
		zooserver = stormConf.get("zkConn").toString();
		ipsearch = new IPSearch(dbJson);
		//		IPSet = ap.GetIPFilter();
		//		map = ap.IPDataBaseMap();
		//		IPs = ap.StartIPs();
		_ipseg = new IDCIPSegment(dbJson);

		//ip过滤
		s = new Slave(zooserver);
		IPSet = s.GetIPFilterForLog();
	}

	public void emitCountingData() {
		//			if(ap.GetIPFilter().size() >0 ) {
		//				IPSet = ap.GetIPFilter();
		//			}
		if(s.GetIPFilterForLog() != null && s.GetIPFilterForLog().size() > 0) {
			IPSet = s.GetIPFilterForLog();
		}

	}

}
